/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.niotcp;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Set;
import java.util.Vector;

import cgl.narada.transport.TransmissionManager;
import cgl.narada.transport.TransmissionReporter;
import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportException;

/**
 * Responsible for sending niotcp packets across the socket connection that is
 * established during initializations. The NIOTCPSenderThread notifies any loss
 * of connections to the NIOTCPLink which in turn proceeds to noitify the
 * NIOTCPLinkFactory.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class NIOTCPSenderThread extends Thread implements TransportDebugFlags,
    TransmissionReporter {
    private SocketChannel sc;

    private NIOTCPLink niotcpLink;

    private int packetTracker = 0;

    private ByteBuffer sizeBuffer;

    private int BUFFER_SIZE = 1024 * 8;

    private int tracker = 0;

    private int bytesReleased = 0;

    private LinkedList<ByteBuffer> sendBuffers;

    private Object syncObject;

    private Selector writeSelector;

    private boolean sleeping = true;

    private int pending = 0;

    private int intraPacketCount = 0;

    private long timeLastSent = 0;

    private Vector bytesLagBuffer;

    private int TIME_SLICE = 10;

    /** Time is in milliseconds */
    private int BYTES_PER_TIMESLICE = BUFFER_SIZE * 2;

    private boolean throttle = false;

    private long ongoingTransfer = 0;

    private int sendBufferSize = 12 * 8192;

    private boolean shutdownThread = false;

    private String moduleName = "NIOTCPSenderThread: ";

    public NIOTCPSenderThread(SocketChannel sc, NIOTCPLink niotcpLink)
        throws Exception {
        this.sc = sc;
        this.niotcpLink = niotcpLink;
        setPriority(MAX_PRIORITY);
        registerTransmissionReporter(niotcpLink);
        sc.configureBlocking(false);
        
        sendBuffers = new LinkedList<ByteBuffer>();
        writeSelector = Selector.open();
        sc.register(writeSelector, SelectionKey.OP_WRITE);
        syncObject = new Object();

        bytesLagBuffer = new Vector();

        sizeBuffer = ByteBuffer.allocate(4);

        sc.socket().setSendBufferSize(sendBufferSize);
    }

    /**
     * This method registers the transmission reporter to the transmission
     * manager.
     */
    private void registerTransmissionReporter(NIOTCPLink niotcpLink) {
        TransmissionManager transmissionManager = niotcpLink
            .getTransportHandler().getTransmissionManager();
        if (transmissionManager != null) {
            transmissionManager.registerTransmissionReporter(niotcpLink
                .getLinkId(), this);
        }
    }

    public void releaseResources() {
        if (NIOTCPSenderThread_Debug)
            System.out.println(moduleName + "Closing " + sc);
        try {
            sc.close();
            shutdownThread = true;
            synchronized (syncObject) {
                syncObject.notify();
            }
            ;
            writeSelector.close();
        } catch (IOException ioe) {
            System.out.println(moduleName + "Problem while releasing resources"
                + ioe);
        }
    }

    private void prepareToSendData(ByteBuffer readyToSend) throws Exception {

        int _willSend = readyToSend.remaining();

        sizeBuffer.clear();
        sizeBuffer.putInt(_willSend);
        sizeBuffer.flip();

        while (sizeBuffer.hasRemaining()) {
            sc.write(sizeBuffer);
        }

        int _actuallySent = 0;

        while (readyToSend.hasRemaining()) {
            _actuallySent += sc.write(readyToSend);
        }

        pending -= _willSend;

        intraPacketCount += _willSend;
        bytesReleased += _willSend;
        tracker += _actuallySent;
    }

    public synchronized void sendData(byte[] dataToSend)
                                                        throws TransportException {
        if (NIOTCPSenderThread_Debug) {
            System.out.println(moduleName + "Sending ..." + dataToSend.length);
        }
        populateSendBuffers(dataToSend);
    }

    private void populateSendBuffers(byte[] dataToSend)
                                                       throws TransportException {

        int dataLength = dataToSend.length;

        try {
            if (dataLength <= 0) {
                System.out.println("\n **Sender: Data length <= 0 *** \n");
                return;
            }
            packetTracker++;

            ByteBuffer wrapBuffer = ByteBuffer.wrap(dataToSend);

            synchronized (sendBuffers) {
                sendBuffers.add(wrapBuffer);
            }
            if (sleeping && (sendBuffers.size() > 0)) {
                synchronized (syncObject) {
                    syncObject.notify();
                }
                ;

            }

        } catch (Exception ioe) {
            String errorInfo = moduleName + "sending data " + ioe;
            System.out.println(errorInfo);
            throw new TransportException(errorInfo);
        }
    }

    public void sendData(byte[] dataToSend, String dataType)
                                                            throws TransportException {
        /** NBNI */
        sendData(dataToSend);
    }

    public void run() {

        try {
            while (true) {
                if (sendBuffers.size() == 0) {
                    /* Thread.sleep(20); continue; */
                    synchronized (syncObject) {
                        sleeping = true;
                        if (NIOTCPSenderThread_Debug) {
                            System.out.println(moduleName + "Pending bytes = ["
                                + pending + "] on link "
                                + niotcpLink.getLinkId());
                        }
                        syncObject.wait();
                        if (shutdownThread) {
                            break;
                        }

                        sleeping = false;
                    }
                    ;
                }

                int status = writeSelector.select();
                if (status == 0) {
                    continue;
                }

                Iterator it = writeSelector.selectedKeys().iterator();

                while (it.hasNext()) {
                    SelectionKey key = (SelectionKey) it.next();

                    if (key.isWritable()) {
                        ByteBuffer bufferToGo;
                        synchronized (sendBuffers) {
                            if (sendBuffers.size() == 0) {
                                break;
                            }
                            bufferToGo = sendBuffers.remove();
                        }

                        int bytesToGo = bufferToGo.capacity();

                        boolean _sent = false;
                        while (!_sent) {
                            if (canReleaseBuffer(bytesToGo)) {
                                prepareToSendData(bufferToGo);
                                _sent = true;
                            } else {
                                Thread.sleep(TIME_SLICE / 2);
                            }/* end if(canRelease) */
                        } /* end while (not sent) */
                        bufferToGo = null;
                    }
                    /** if key.isWritable() */
                    it.remove();
                }/* end while (it.hasNext()) */

            } /* end while(true) */
        } catch (Exception ie) {
            ie.printStackTrace();
        }
        /* end try-catch */

        System.out.println(moduleName + "Socket Connection to ["
            + niotcpLink.getLinkId() + "] broken");

    }/* end run */

    private boolean canReleaseBuffer(int numBytes) {
        if (!throttle) {
            return true;
        }

        if (timeLastSent == 0) {
            timeLastSent = System.currentTimeMillis();
            return true;
            /** This is true only for the first time a message is sent */
        }

        if (bytesLagBuffer.size() == 0) {
            System.out.println(moduleName
                + "ByteLabBuffer.size() = 0 ... Problem");
            return true;
        }

        int projectedLag = (int) (System.currentTimeMillis() - timeLastSent);
        if (NIOTCPSenderThread_Debug) {
            System.out.println(moduleName + "Projected Lag= " + projectedLag);
        }

        if (projectedLag >= TIME_SLICE) {
            /**
             * Since no messages were sent in the previous time slice, we can
             * send this message. We should of course clear the bytesLagBuffer
             * since the entries are now of no use to us
             */
            bytesLagBuffer.clear();
            return true;
        }

        /**
         * We now need to eliminate bytesLagEntries that took place TIME_SLICE
         * milliseconds before the current instant
         */
        int cummulativeLag = projectedLag;
        boolean clearEntries = false;
        for (int i = bytesLagBuffer.size() - 1; i >= 0; i--) {

            int _lag = ((BytesAndTimeLag) bytesLagBuffer.elementAt(i))
                .getTimeLag();
            cummulativeLag += _lag;
            if (cummulativeLag >= TIME_SLICE) {
                /**
                 * Every entry after this one needs to be invalidated and
                 * removed since we are interested in only what happened in the
                 * TIME_SLICE before NOW.
                 */
                if (NIOTCPSenderThread_Debug) {
                    System.out.println(moduleName + "Clearing [" + i
                        + "] enrties from the bytesLagBuffer");
                }
                for (int j = 0; j < i; j++) {
                    bytesLagBuffer.removeElementAt(0);
                }
                break;
            }
            /** if cummulative > TIME_SLICE */
        } /* end (for) */

        /**
         * Check to see if we have reached the number of bytes_per_timeslice
         * threshold. If we have don't send otherwise send.
         */
        int _byteSlice = numBytes;
        for (int i = 0; i < bytesLagBuffer.size(); i++) {
            int _bytesSent = ((BytesAndTimeLag) bytesLagBuffer.elementAt(i))
                .getBytesSent();
            _byteSlice += _bytesSent;
        }

        if (_byteSlice > BYTES_PER_TIMESLICE) {
            if (NIOTCPSenderThread_Debug) {
                System.out.println(moduleName
                    + "Bytes per time slice would be " + _byteSlice
                    + ", so not sending now");
            }
            return false;
        }

        return true;

    }/* end canReleaseBuffer() */

    public boolean hasPendingTransfer() {
        long currentTime = System.currentTimeMillis();
        long timeAfterLastTransfer = currentTime - timeLastSent;
        if (timeAfterLastTransfer < 100) {
            System.out.println(moduleName + "There was a transfer ("
                + timeAfterLastTransfer + ") msec ago");
            return true;
        }

        if (sendBuffers.size() != 0) {
            return true;
        }
        return false;
    }

    public long getSizeOfOngoingTransfer() {
        int numOfBuffers = sendBuffers.size();
        ongoingTransfer = numOfBuffers * BUFFER_SIZE;
        return ongoingTransfer;
    }

}
